package org.example.http;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

public class DataHandler {

    private static final Logger LOG = LoggerFactory.getLogger(DataHandler.class);

    public static DruidDataSource mysqlsource;

    public static DruidDataSource mssqlsource;

    public static DruidDataSource oraclesource;

    public static DruidDataSource dorisDataSource;

    private final static int MAX_ACTIVE = 300;

    private static AtomicInteger userCount = new AtomicInteger(0);

    static {

        mysqlsource = new DruidDataSource();
        mysqlsource.setUrl("jdbc:mysql://192.168.209.246:3306/test_flinkx?useSSL=false");
        mysqlsource.setUsername("root");
        mysqlsource.setPassword("Infra5@Gep0int");
        mysqlsource.setMaxActive(MAX_ACTIVE);
        mysqlsource.setLogAbandoned(true);
        mysqlsource.setRemoveAbandoned(true);
        mysqlsource.setRemoveAbandonedTimeout(180);

        mssqlsource = new DruidDataSource();
        mssqlsource.setUrl("jdbc:sqlserver://192.168.186.7:1433;database=epoint");
        mssqlsource.setUsername("dba_epoint");
        mssqlsource.setPassword("11111");
        mssqlsource.setMaxActive(MAX_ACTIVE);
        mssqlsource.setLogAbandoned(true);
        mssqlsource.setRemoveAbandoned(true);
        mssqlsource.setRemoveAbandonedTimeout(180);

        oraclesource = new DruidDataSource();
        oraclesource.setUrl("jdbc:oracle:thin:@192.168.186.36:1521:orcl");
        oraclesource.setUsername("epoint");
        oraclesource.setPassword("11111");
        oraclesource.setMaxActive(MAX_ACTIVE);
        oraclesource.setLogAbandoned(true);
        oraclesource.setRemoveAbandoned(true);
        oraclesource.setRemoveAbandonedTimeout(180);

        dorisDataSource = new DruidDataSource();
        dorisDataSource.setUrl("jdbc:mysql://192.168.186.66:9030/dorisdb_test?useSSL=false");
        dorisDataSource.setUsername("test_user");
        dorisDataSource.setPassword("11111");
        dorisDataSource.setMaxActive(MAX_ACTIVE);
        dorisDataSource.setLogAbandoned(true);
        dorisDataSource.setRemoveAbandoned(true);
        dorisDataSource.setRemoveAbandonedTimeout(180);
    }


    public static void mysqlBatch(int count, int batchSize,
                                  String tableName, List<String> fields) {

        if (StringUtils.isBlank(tableName)) {
            LOG.error("tablename can not empty");
            return;
        }
        if (fields == null || fields.size() == 0) {
            LOG.error("fields can not empty and length must be > 0");
            return;
        }

        String fieldsStr = StringUtils.join(fields, ",");
        List<String> parameterList = new ArrayList<>();
        fields.forEach(item->{
            parameterList.add("?");
        });
        String sql = String.format("insert into %s(%s) values(%s)", tableName, fieldsStr, StringUtils.join(parameterList));
        // String sql = "insert into jizy_flink (id, createtime, updatetime) values(?, ?, ?)";
        try {

            for (int i = 0; i < count; i++) {
                DruidPooledConnection connection = mysqlsource.getConnection();
                PreparedStatement statement = connection.prepareStatement(sql);
                connection.setAutoCommit(false);
                System.out.println("导入开始" + i + "开始=====>");
                for (int j = 0; j < batchSize; j++) {
                    statement.setObject(1, UUID.randomUUID().toString());
                    statement.setObject(2, new Date());
                    statement.setObject(3, new Date());
//                    statement.setObject(4, "用户" + (i * batchSize + j));
                    statement.addBatch();
                }
                statement.executeBatch();
                statement.clearBatch();
                System.out.println("导入结束" + i + "结束=====>");
                connection.commit();
                statement.close();
                connection.close();
            }

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public static void sqlserBatch(int count, int batchSize) {

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        String sql = "insert into jizy_flink (id, createtime, updatetime, \"user\") values(?, ?, ?, ?)";
        try {

            for (int i = 0; i < count; i++) {
                DruidPooledConnection connection = mssqlsource.getConnection();
                PreparedStatement statement = connection.prepareStatement(sql);
                connection.setAutoCommit(false);
                System.out.println("导入开始" + i + "开始=====>");
                for (int j = 0; j < batchSize; j++) {
                    statement.setObject(1, UUID.randomUUID().toString());
                    statement.setObject(2, sdf.format(new Date()));
                    statement.setObject(3, sdf.format(new Date()));
                    statement.setObject(4, "用户sqlserver" + (i * batchSize + j));
                    statement.addBatch();
                }
                statement.executeBatch();
                statement.clearBatch();
                System.out.println("导入结束" + i + "结束=====>");
                connection.commit();
                statement.close();
                connection.close();
            }

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public static void oracleBatch(int count, int batchSize) {

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        String sql = "insert into EPOINT.JIZY_FLINK (id, createtime, updatetime, username) values(?, ?, ?, ?)";
        try {

            for (int i = 0; i < count; i++) {
                DruidPooledConnection connection = oraclesource.getConnection();
                PreparedStatement statement = connection.prepareStatement(sql);
                connection.setAutoCommit(false);
                System.out.println("导入开始" + i + "开始=====>");
                for (int j = 0; j < batchSize; j++) {
                    statement.setObject(1, UUID.randomUUID().toString());
                    statement.setObject(2, sdf.format(new Date()));
                    statement.setObject(3, sdf.format(new Date()));
                    statement.setObject(4, "用户oracle" + (i * batchSize + j));
                    statement.addBatch();
                }
                statement.executeBatch();
                statement.clearBatch();
                System.out.println("导入结束" + i + "结束=====>");
                connection.commit();
                statement.close();
                connection.close();
            }

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public static void insertSiteUser(int count, int batchSize,
                                  String tableName, List<String> fields) {

        if (StringUtils.isBlank(tableName)) {
            LOG.error("tablename can not empty");
            return;
        }
        if (fields == null || fields.size() == 0) {
            LOG.error("fields can not empty and length must be > 0");
            return;
        }

        String fieldsStr = StringUtils.join(fields, ",");
        List<String> parameterList = new ArrayList<>();

        int fieldsLen = fields.size();
        for (int i = 0; i < fieldsLen; i++) {
            parameterList.add("?");
        }
        String sql = String.format("insert into %s(%s) values(%s)", tableName, fieldsStr, StringUtils.join(parameterList, ","));
        try {

            for (int i = 0; i < count; i++) {
                DruidPooledConnection connection = mysqlsource.getConnection();
                PreparedStatement statement = connection.prepareStatement(sql);
                connection.setAutoCommit(false);
                System.out.println("导入开始" + i + "开始=====>");
                for (int j = 0; j < batchSize; j++) {
                    // TODO 以10000用户来计数
                    int userIndex = userCount.incrementAndGet() % 10000;
                    statement.setObject(1, "用户" + userIndex);
                    statement.setObject(2, UUID.randomUUID().toString());
                    statement.setObject(3, String.format("%018d", userIndex));
                    statement.setObject(4, String.format("%011d", userIndex));
                    statement.setObject(5, "215600");
                    statement.addBatch();
                }
                statement.executeBatch();
                statement.clearBatch();
                System.out.println("导入结束" + i + "结束=====>");
                connection.commit();
                statement.close();
                connection.close();
            }

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public static void insertSiteAccess(int count, int batchSize,
                                      String tableName, List<String> fields) {

        if (StringUtils.isBlank(tableName)) {
            LOG.error("tablename can not empty");
            return;
        }
        if (fields == null || fields.size() == 0) {
            LOG.error("fields can not empty and length must be > 0");
            return;
        }

        List<Map<String, Object>> list = SiteUserService.siteUserList;
        int usersLen = list.size();
        Random random = new Random();

        String fieldsStr = StringUtils.join(fields, ",");
        List<String> parameterList = new ArrayList<>();

        int fieldsLen = fields.size();
        for (int i = 0; i < fieldsLen; i++) {
            parameterList.add("?");
        }
        String sql = String.format("insert into %s(%s) values(%s)", tableName, fieldsStr, StringUtils.join(parameterList, ","));
        try {

            for (int i = 0; i < count; i++) {
                DruidPooledConnection connection = mysqlsource.getConnection();
                PreparedStatement statement = connection.prepareStatement(sql);
                connection.setAutoCommit(false);
                System.out.println("导入开始" + i + "开始=====>");
                for (int j = 0; j < batchSize; j++) {
                    // TODO 以10000用户来计数
                    Map<String, Object> user = list.get(random.nextInt(usersLen));
                    statement.setObject(1, new Date());
                    statement.setObject(2, 0);
                    statement.setObject(3, 0);
                    statement.setObject(4, user.get("user_name"));
                    statement.setObject(5, user.get("user_id"));
                    statement.setObject(6, 100);
                    statement.addBatch();
                }
                statement.executeBatch();
                statement.clearBatch();
                System.out.println("导入结束" + i + "结束=====>");
                connection.commit();
                statement.close();
                connection.close();
            }

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    /**
     * 测试查询
     */
    public static int querySiteAccess() {
        DruidPooledConnection connection = null;
        int result = -1;
        try {
            connection = dorisDataSource.getConnection();

            PreparedStatement statement = connection.prepareStatement("select count(user_id) as total from site_access");
            ResultSet resultSet = statement.executeQuery();
            if (resultSet.next()) {
                result = resultSet.getInt("total");
            }
            statement.close();
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        return result;
    }

    public static void query1() {
        DruidPooledConnection connection = null;
        int result = -1;
        try {
            connection = dorisDataSource.getConnection();

            PreparedStatement statement = connection.prepareStatement("select sum(lo_revenue) as revenue\n" +
                    "from lineorder join dates on lo_orderdate = d_datekey\n" +
                    "where d_year = 1993 and lo_discount between 1 and 3 and lo_quantity < 25");
            ResultSet resultSet = statement.executeQuery();
            if (resultSet.next()) {
                System.out.println("result = " + resultSet.getString("revenue"));
                // result = resultSet.getInt("revenue");
            }
            statement.close();
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void query2() {
        DruidPooledConnection connection = null;
        int result = -1;
        try {
            connection = dorisDataSource.getConnection();

            PreparedStatement statement = connection.prepareStatement("select sum(lo_revenue) as revenue\n" +
                    "from lineorder\n" +
                    "join dates on lo_orderdate = d_datekey\n" +
                    "where d_yearmonthnum = 199401\n" +
                    "and lo_discount between 4 and 6\n" +
                    "and lo_quantity between 26 and 35");
            ResultSet resultSet = statement.executeQuery();
            if (resultSet.next()) {
                System.out.println("result = " + resultSet.getString("revenue"));
            }
            statement.close();
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void query3() {
        DruidPooledConnection connection = null;
        int result = -1;
        try {
            connection = dorisDataSource.getConnection();

            PreparedStatement statement = connection.prepareStatement("select sum(lo_revenue) as revenue\n" +
                    "from lineorder\n" +
                    "join dates on lo_orderdate = d_datekey\n" +
                    "where d_weeknuminyear = 6 and d_year = 1994\n" +
                    "and lo_discount between 5 and 7\n" +
                    "and lo_quantity between 26 and 35");
            ResultSet resultSet = statement.executeQuery();
            if (resultSet.next()) {
                System.out.println("result = " + resultSet.getString("revenue"));
            }
            statement.close();
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void query4() {
        DruidPooledConnection connection = null;
        int result = -1;
        try {
            connection = dorisDataSource.getConnection();

            PreparedStatement statement = connection.prepareStatement("select sum(lo_revenue) as lo_revenue, d_year, p_brand\n" +
                    "from lineorder\n" +
                    "join dates on lo_orderdate = d_datekey\n" +
                    "join part on lo_partkey = p_partkey\n" +
                    "join supplier on lo_suppkey = s_suppkey\n" +
                    "where p_category = 'MFGR#12' and s_region = 'AMERICA'\n" +
                    "group by d_year, p_brand\n" +
                    "order by d_year, p_brand");
            ResultSet resultSet = statement.executeQuery();
            statement.close();
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
